package gd1;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import util.CDCUtil;
import util.KafkaUtil;

/**
 * CDC数据采集处理类
 * 功能：从MySQL CDC数据源读取变更数据，过滤有效数据后写入Kafka ODS层
 */
public class Ods_CDC {
    public static void main(String[] args) throws Exception {
        // 1. 创建Flink流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); // 设置并行度为1（生产环境应根据实际情况调整）

        // 2. 从MySQL CDC数据源获取数据流
        // 使用CDCUtil工具类获取MySQL变更数据，day0906是数据源标识
        DataStreamSource<String> app_user_shop = CDCUtil.getMysqlSource(env, "day0906");

        // 3. 打印原始数据（调试用）
        app_user_shop.print();

        // 4. 执行数据过滤处理
        filter(app_user_shop);

        // 5. 启动作业
        env.execute("CDC_Data_Processing_Job"); // 作业名称
    }

    /**
     * 数据过滤处理方法
     * @param app_user_shop 原始CDC数据流
     */
    private static void filter(DataStreamSource<String> app_user_shop) {
        // 1. 过滤无效数据
        SingleOutputStreamOperator<String> after = app_user_shop.filter(
                new FilterFunction<String>() {
                    @Override
                    public boolean filter(String s) throws Exception {
                        // 1.1 检查是否为有效JSON
                        boolean valid = JSON.isValid(s);
                        if (!valid) {
                            return false;
                        }

                        // 1.2 检查是否包含after字段（实际变更数据）
                        if (JSON.parseObject(s).getString("after") == null) {
                            return false;
                        }

                        return true; // 同时满足JSON有效且包含after字段的数据才会被保留
                    }
                }
        );
        // 2. 打印过滤后的数据（调试用）
        after.print();

        // 3. 将有效数据写入Kafka ODS层
        after.addSink(KafkaUtil.getKafkaSink("ods_day0906"));
    }
}